package cn.ac.iscas.service.source;

import cn.ac.iscas.base.utils.ParameterUtils;
import cn.ac.iscas.base.utils.SpringUtil;
import cn.ac.iscas.domain.source.FlinkSourceColumn;
import cn.ac.iscas.service.queue.SqlExecuteQueueService;
import cn.ac.iscas.service.queue.TableQueueService;
import com.iscas.datasong.lib.common.DataSongException;
import com.iscas.datasong.lib.util.DataSongJsonUtils;
import org.springframework.stereotype.Service;

import java.io.File;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import static cn.ac.iscas.base.constant.FileSystemConnectConstant.*;

/**
 * @author LJian
 * @version 1.0
 * @description: TODO
 * @date 2022/6/6 10:15
 */
@Service
public class CsvSource {
    private final static String CONNECTOR_TYPE_VAULE = "filesystem";
    private final static String CONNECTOR_FORMAT_VALUE = "csv";
    private String componentId;
    private String jobId;

    public CsvSource(String jobId, String taskId) {
        this.jobId = jobId;
        this.componentId = taskId;
    }

    public CsvSource() {

    }

    public String getSql(Map<String, Object> parameters) throws DataSongException {
        String path = ParameterUtils.getRequiredString(parameters, "path");
        String delmeter = ParameterUtils.getOrDefaultString(parameters, "fieldDelimiter", ",");
        boolean isSkipFirstLine = ParameterUtils.getOrDefaultBoolean(parameters, "isSkipFirstLine", false);
        List<Map<String, Object>> columns = ParameterUtils.getRequiredList(parameters, "columns");
        String table = path.substring(path.replace("\\", File.pathSeparator).lastIndexOf("/")+1).split("\\.")[0];


        StringBuffer tableSchemaBuf = new StringBuffer(String.format("CREATE TABLE %s ( ", table));
        for (Map<String, Object> column : columns) {
            tableSchemaBuf.append(String.format("%s %s, ", ParameterUtils.getRequiredString(column, "name"), ParameterUtils.getRequiredString(column, "type")));
        }
        String tableSchemaString = tableSchemaBuf.toString();
        if (tableSchemaString.trim().endsWith(",")) {
            tableSchemaString = tableSchemaString.substring(0, tableSchemaString.lastIndexOf(","));
        }
        tableSchemaString += ") ";
        StringBuffer withBuf = new StringBuffer("WITH (")
                .append(String.format("'%s' = '%s'", CONNECTOR_TYPE_KEY, CONNECTOR_TYPE_VAULE))
                .append(String.format(",'%s' = '%s'", CONNECTOR_PATH_KEY, path))
                .append(String.format(",'%s' = '%s'", CONNECTOR_FORMAT_KEY, CONNECTOR_FORMAT_VALUE));

        String withStr = withBuf.append(")").toString();

        TableQueueService tableQueueService = SpringUtil.getBean(TableQueueService.class);
        List<FlinkSourceColumn> columnList = ParameterUtils.getRequiredList(parameters, "columns");
//        List<FlinkSourceColumn> columnList = DataSongJsonUtils.fromJson(columnsJson, List.class);
        ConcurrentHashMap<String, List<FlinkSourceColumn>> tableColumns = new ConcurrentHashMap<>();
        tableColumns.put(table, columnList);
        tableQueueService.put(this.jobId, this.componentId, tableColumns);
        return tableSchemaString + withStr;
    }
}
